package felix.rule.service.actors.sharde;

import felix.actor.core.TActorCtx;
import felix.rule.data.msg.EngineMsg;
import felix.rule.data.msg.RuleNodeException;
import felix.rule.data.plugin.ComponentLifecycleState;
import felix.rule.service.actors.ActorSystemContext;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public abstract class ComponentMsgProcessor extends AbstractContextAwareMsgProcessor {

    protected final String id;
    protected final String ruleId;
    protected ComponentLifecycleState state;

    protected ComponentMsgProcessor(ActorSystemContext systemContext, String ruleId, String id) {
        super(systemContext);
        this.ruleId = ruleId;
        this.id = id;
    }

    public abstract String getComponentName();

    public abstract void start(TActorCtx context) throws Exception;

    public abstract void stop(TActorCtx context) throws Exception;

//    public abstract void onPartitionChangeMsg(PartitionChangeMsg msg) throws Exception;

    public void onCreated(TActorCtx context) throws Exception {
        start(context);
    }

    public void onUpdate(TActorCtx context) throws Exception {
        restart(context);
    }

    public void onActivate(TActorCtx context) throws Exception {
        restart(context);
    }

    public void onSuspend(TActorCtx context) throws Exception {
        stop(context);
    }

    public void onStop(TActorCtx context) throws Exception {
        stop(context);
    }

    private void restart(TActorCtx context) throws Exception {
        stop(context);
        start(context);
    }

//    public void scheduleStatsPersistTick(TActorCtx context, long statsPersistFrequency) {
//        schedulePeriodicMsgWithDelay(context, new StatsPersistTick(), statsPersistFrequency, statsPersistFrequency);
//    }

    protected void checkActive(EngineMsg msg) throws RuleNodeException {
        if (state != ComponentLifecycleState.ACTIVE) {
            log.debug("Component is not active. Current state [{}] for processor [{}] tenant [{}]", state, id, ruleId);
            RuleNodeException ruleNodeException = getInactiveException();
            if (msg != null) {
                msg.getCallback().onFailure(ruleNodeException);
            }
            throw ruleNodeException;
        }
    }

    abstract protected RuleNodeException getInactiveException();

}
